package cn.fw.community.adapter.facade;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * @author cn
 * @version 1.0
 * @description
 * @date 2022/9/5 17:26
 */
@Slf4j
public class ThreadPoolUtils {


  private static <T> CompletableFuture<Void> runConsumer(AtomicBoolean endExec, T t, Consumer<T> consumer, Executor executor) {
    Runnable runnable = buildRunnable(endExec, t, consumer);
    if (executor == null) {
      return CompletableFuture.runAsync(runnable);
    }
    return CompletableFuture.runAsync(runnable, executor);
  }

  private static <T> Runnable buildRunnable(AtomicBoolean endExec, T t, Consumer<T> consumer) {
    return () -> {
      if (!endExec.get()) {
        consumer.accept(t);
        System.out.println("我执行完了:" + t);
      } else {
        System.out.println("我并没有执行，就结束了:" + t);
      }
    };
  }

  private static <T, R> R buildRunnable(AtomicBoolean endExec, T t, Function<T, R> consumer) {
    if (!endExec.get()) {
      R apply = consumer.apply(t);
      System.out.println("我执行完了:" + apply);
      return apply;
    } else {
      System.out.println("我并没有执行，就结束了:" + t);
      return null;
    }
  }

  private static <T, R> CompletableFuture<R> supplyConsumer(AtomicBoolean endExec, T t, Function<T, R> consumer, Executor executor) {
    Supplier<R> r = () -> buildRunnable(endExec, t, consumer);
    if (executor == null) {
      return CompletableFuture.supplyAsync(r);
    }
    return CompletableFuture.supplyAsync(r, executor);
  }

  /**
   * 自动执行，只要其中一个执行完成，其他的就不管状态(失败,或者是丢弃,或者是终断)
   *
   * @param task     任务列表
   * @param consumer 消费操作
   * @param executor
   * @param <T>
   */
  public static <T> void anyOfRunAsync(Collection<T> task, Consumer<T> consumer, Executor executor) {
    List<CompletableFuture<?>> result = new ArrayList<>(task.size());
    //终止执行，默认为false
    AtomicBoolean endExec = new AtomicBoolean();
    for (T t : task) {
      result.add(runConsumer(endExec, t, consumer, executor));
    }
    CompletableFuture.anyOf(result.toArray(new CompletableFuture[0])).join();
    endExec.set(true);
  }

  public static <T, R> void anyOfRunAsync(Collection<T> task, Function<T, R> consumer,
                                          Predicate<R> predicate, Executor executor) {
    List<CompletableFuture<R>> result = new ArrayList<>(task.size());
    //终止执行，默认为false
    AtomicBoolean endExec = new AtomicBoolean();
    for (T t : task) {
      result.add(supplyConsumer(endExec, t, consumer, executor));
    }
    for (CompletableFuture<R> rCompletableFuture : result) {
      predicateEndAction(rCompletableFuture, endExec, predicate, executor);
    }
  }

  private static <R> void predicateEndAction(CompletableFuture<R> rCompletableFuture,
                                             final AtomicBoolean endFlag, Predicate<R> predicate, Executor executor) {
    if (executor == null) {
      executor = ForkJoinPool.commonPool();
    }
    executor.execute(() -> {
      if (endFlag.get()) {
        return;
      }
      rCompletableFuture.join();
      R r = null;
      try {
        r = rCompletableFuture.get();
      } catch (Exception e) {
        log.info("interrupted", e);
        Thread.currentThread().interrupt();
      }
      endFlag.compareAndSet(false, predicate.test(r));
    });
  }


  public static void main(String[] args) {
    ExecutorService executorService = Executors.newFixedThreadPool(1000);
    List<Integer> collect = IntStream.range(1000, 3000).boxed().collect(Collectors.toList());
    anyOfRunAsync(collect, (e) -> {
      int s = e;
      try {
        Thread.sleep((s % 200) == 0 ? s * 10L : s * 5L);
      } catch (InterruptedException ex) {
      }
      return s;
    }, result -> (result % 1200) == 0, executorService);
//    executorService.shutdown();
  }
}
